package com.shujia.spark.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo2DataSource {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Source")
      .master("local")
      .getOrCreate()

    /**
      * 读取json格式的数据
      *json 中自带列名和列的类型，spark读取过程
      * 自动解析列名和列的类型
      */
    val jsonDF: DataFrame = spark.read.format("json").load("data/students.json")
    /**
      * 读取csv格式的数据
      */
    val csvDF: DataFrame = spark
      .read
      .format("csv")
      //指定列分隔方式
      .option("sep", ",")
      .schema("id string,name string,age int,gender string,clazz string")
      .load("data/students.txt")

    /**
      * 保存为parquet格式
      */

    //csvDF.write.mode(SaveMode.Overwrite).parquet("data/parquet")

    /**
      * 读取parquet格式的数据
      *
      * 1、自带列名
      * 2、数据会压缩(解压和压缩会花费时间，压缩比一般是5倍)
      * 3、可以兼容hive
      *
      */
    val parquetDF: DataFrame = spark.read.format("parquet").load("data/parquet")

    parquetDF.printSchema()
    parquetDF.show()

    //先保存为一个orc格式的文件
    //csvDF.write.mode(SaveMode.Overwrite).orc("data/orc")

    /**
      * 读取orc格式
      * 1、自带列名
      * 2、数据会压缩 (压缩和解压需要时间，压缩比一般在5倍左右)
      * 3、可以兼容hive
      */

    val orcDF: DataFrame = spark
      .read
      .format("orc")
      .load("data/orc")
    orcDF.show()
    orcDF.printSchema()

    /**
      * 用jdbc读取mysql中的数据
      */
    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306")
      .option("dbtable","student.student")
      .option("user", "root")
      .option("password", "123456")
      .load()
    jdbcDF.show()

    /**
      * 将数据保存在MySQL
      *
      */
    orcDF.write
      .format("jdbc")
      .mode(SaveMode.Overwrite)
      .option("url","jdbc:mysql://master:3306?useUnicode=true&characterEncoding=utf-8")
      .option("dbtable","student.student1")
      .option("user","root")
      .option("password","123456")
      .save()
  }
}
